package com.jiayuan.cn.energy.test;

import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Optional;

public class RemoteFlink {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("192.168.110.88", 8081);
        StreamTableEnvironment stEnv = StreamTableEnvironment.create(env);
        String kafkaFK = """
                CREATE TABLE csv_source (
                                 user_id INT,
                                 product STRING,
                                 order_amount DOUBLE
                                ) WITH (
                                 'connector' = 'filesystem',
                                 'path' = 'file:///path/input.csv',
                                 'format' = 'csv'
                                )
                """;
        String mysqlFK = """
                CREATE TABLE mysql_sum (
                        user_id INT,
                        product STRING,
                        all_amount DOUBLE,
                        PRIMARY KEY (user_id) NOT ENFORCED
                    ) WITH (
                        'connector'  = 'jdbc',
                        'url'        = 'jdbc:mysql://192.168.110.88:3306/helm',
                        'table-name' = 'mysql_sum',
                        'driver'     = 'com.mysql.cj.jdbc.Driver',
                        'username'   = 'helm',
                        'password'   = 'db-999'
                    )
                """;

        stEnv.executeSql(kafkaFK);
        stEnv.executeSql(mysqlFK);
        TableResult tableResult = stEnv.executeSql("INSERT INTO mysql_sum  SELECT user_id,product, order_amount as total_amount FROM csv_source");
        //获取任务id
        Optional<JobClient> jobClient = tableResult.getJobClient();
        JobClient jobClient1 = jobClient.get();
        JobID jobID = jobClient1.getJobID();
        System.out.println(jobID);

    }
}
